package com.kafka_java;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.Callback;

import java.util.Properties;

public class Producer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "8.130.34.5:32000");
        props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("acks", "all");
        props.setProperty("retries", "3"); // 设置重试次数
        props.setProperty("linger.ms", "1"); // 设置延迟
        props.setProperty("buffer.memory", "33554432"); // 设置缓冲区大小
        props.setProperty("request.timeout.ms", "1000"); // 添加请求超时时间设置

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> message = new ProducerRecord<>("abcx", Integer.toString(i), "abc" + i);
                producer.send(message, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.err.println("Failed to send message with exception: " + exception);
                            exception.printStackTrace(); // 打印异常详细信息
                        } else {
                            System.out.println("Sent message to topic " + metadata.topic() + " partition " + metadata.partition() + " at offset " + metadata.offset());
                        }
                    }
                });
                Thread.sleep(100); // 控制发送速率
            }
        } catch (Exception e) {
            System.err.println("Exception caught in producer: " + e.getMessage());
            e.printStackTrace(); // 捕获异常时打印详细信息
        }
    }
}
